package com.dahua.messagesyncsh.rabbitmq;

import com.dahua.messagesyncsh.rabbitmq.enums.Rabbitmq;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.util.ErrorHandler;

import java.util.Properties;

@Slf4j
@Configuration
public class RabbitmqConfig {
    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        return new RabbitAdmin(connectionFactory);
    }

    @Value("${spring.rabbitmq.host}")
    private String host;

    @Value("${spring.rabbitmq.port}")
    private int port;

    @Value("${spring.rabbitmq.username}")
    private String username;

    @Value("${spring.rabbitmq.password}")
    private String password;

    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory factory = new CachingConnectionFactory();
        factory.setHost(host);
        factory.setPort(port);
        factory.setUsername(username);
        factory.setPassword(password);
        return factory;
    }

    @Bean
    @Primary
    public RabbitTemplate RabbitTemplate(ConnectionFactory connectionFactory) {
        return new RabbitTemplate(connectionFactory);
    }

    /*  manual injects bean into IOC for control consumer*/
    public SimpleMessageListenerContainer messageListenerContainer(Properties properties,
                                                                   MessageListener messageListener,
                                                                   ErrorHandler errorHandler,
                                                                   PlatformTransactionManager transactionManager) {

        ConnectionFactory factory = connectionFactory(properties);

        if (factory == null){
            return null;
        }

        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(factory);
        container.setQueueNames(properties.getProperty(Rabbitmq.QUEUE.getName()));
        /*  message handler*/
        container.setMessageListener(messageListener);

        try {
            setContainer(
                    container,
                    properties,
                    errorHandler,
                    transactionManager
            );
        } catch (Exception e) {
            log.error("【rabbitmq】：container设置错误，{}", e.toString());
            return null;
        }

        return container;
    }

    public void setContainer(SimpleMessageListenerContainer container,
                             Properties properties,
                             ErrorHandler errorHandler,
                             PlatformTransactionManager transactionManager) {
        if (errorHandler != null) {
            container.setErrorHandler(errorHandler);
        }

        /*  enable transaction or not*/
        if (transactionManager != null) {
            container.setChannelTransacted(true);
            container.setTransactionManager(transactionManager);
        }

        container.setDefaultRequeueRejected(properties.contains(Rabbitmq.DEFAULT_REQUEUE_REJECTED.getName()));

        /*  auto ack or manual*/
        if (properties.containsKey(Rabbitmq.MANUAL_ACK.getName())) {
            container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        } else {
            container.setAcknowledgeMode(AcknowledgeMode.AUTO);
        }

        if (properties.containsKey(Rabbitmq.PRE_FETCH_COUNT.getName())) {
            container.setPrefetchCount(Integer.parseInt(properties.getProperty(Rabbitmq.PRE_FETCH_COUNT.getName())));
        }

        if (properties.containsKey(Rabbitmq.CONCURRENT_CONSUMERS.getName())) {
            container.setConcurrentConsumers(Integer.parseInt(properties.getProperty(Rabbitmq.CONCURRENT_CONSUMERS.getName())));
        }

        if (properties.containsKey(Rabbitmq.MAX_CONCURRENT_CONSUMERS.getName())) {
            container.setMaxConcurrentConsumers(Integer.parseInt(properties.getProperty(Rabbitmq.MAX_CONCURRENT_CONSUMERS.getName())));
        }

        if (properties.containsKey(Rabbitmq.ENABLE_BATCH.getName())) {
            if (Boolean.parseBoolean(properties.getProperty(Rabbitmq.ENABLE_BATCH.getName()))) {
                container.setConsumerBatchEnabled(true);
                if (properties.containsKey(Rabbitmq.BATCH_SIZE.getName())) {
                    container.setBatchSize(Integer.parseInt(properties.getProperty(Rabbitmq.BATCH_SIZE.getName())));
                }
            }
        }

    }

    public ConnectionFactory connectionFactory(Properties properties) {
        /*  verification*/
        if (properties == null ||
                properties.isEmpty() ||
                !properties.containsKey(Rabbitmq.HOST.getName()) ||
                !properties.containsKey(Rabbitmq.PORT.getName()) ||
                !properties.containsKey(Rabbitmq.USERNAME.getName()) ||
                !properties.containsKey(Rabbitmq.PASSWORD.getName()))
        {
            log.error("【rabbitmq】：properties配置缺失");
            return null;
        }

        CachingConnectionFactory factory = new CachingConnectionFactory();
        try {
            factory.setHost(properties.getProperty(Rabbitmq.HOST.getName()));
            factory.setPort(Integer.parseInt(properties.getProperty(Rabbitmq.PORT.getName())));
            factory.setUsername(properties.getProperty(Rabbitmq.USERNAME.getName()));
            factory.setPassword(properties.getProperty(Rabbitmq.PASSWORD.getName()));
        } catch (Exception e) {
            log.error("【rabbitmq】：properties配置错误，{}", e.toString());
            return null;
        }
        return factory;
    }
}
